package org.zjt.spark.book

import java.text.SimpleDateFormat
import java.util.Date
import java.util.concurrent.locks.ReentrantLock

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
  * DESC    马尔科夫预测
  * 结果为：（(开始状态：结束状态)，统计值)作为马尔科夫的输入依据。感觉马尔科夫生成概率矩阵。
  *
  * @author
  * @create 2017-06-23 上午10:47
  **/
object ProductForecaster extends App {


  var sparkConf = new SparkConf().setMaster("local[2]").setAppName("ProductForecaster")
  val sc = new SparkContext(sparkConf)

  /**
    * SimpleDateFormat线程不安全隐患  SimpleDateFormat
    */
  val df = MyDateParser
  val broadcastBateFormat = sc.broadcast(df)


  sc.setCheckpointDir("/Users/zhangjuntao/IdeaProjects/myproject/hw-bigdata/scala-demo/src/main/resource/checkpoint")
  val rdd = sc.textFile("/Users/zhangjuntao/IdeaProjects/myproject/hw-bigdata/scala-demo/src/main/resource/(sample)sam_tianchi_2014002_rec_tmall_log2.csv").coalesce(2).map {
    line => {
      val array = line.split(",")
      (array(1), (array(3), array(2)))
    }
  }.groupByKey().map {
    a => {
      val v = a._2.toArray.sortWith {
        (a, b) => {
          val dateFormat = broadcastBateFormat.value
          if (a._1.length == 0 || b._1.length == 0) false
          else dateFormat.parse(a._1.toString).getTime < dateFormat.parse(b._1.toString).getTime
        }
      }
      println(a._1 + ":" + v.mkString(","))
      (a._1, v)
    }
  }.persist().map {
    line => {
      var array = ArrayBuffer[String]()
      val values = line._2
      if (values.length >= 2) {
        var i = 1
        while (i < values.length) {
          val preDate = values(i - 1)._1
          val preAmount = values(i - 1)._2
          val nowDate = values(i)._1
          val nowAmount = values(i)._2
          i += 1
          val dateFormat = broadcastBateFormat.value
          val dayDif = (dateFormat.parse(nowDate).getTime - dateFormat.parse(preDate).getTime) / (1000 * 3600 * 24)
          val amountDif = nowAmount.toDouble - preAmount.toDouble

          var dd: String = ""
          if (dayDif < 30) {
            dd = "S"
          } else if (dayDif < 60) {
            dd = "M"
          } else {
            dd = "L"
          }
          var ad: String = ""
          if (preAmount.toDouble < nowAmount.toDouble * 0.9) {
            ad = "L"
          } else if (preAmount.toDouble < nowAmount.toDouble * 1.1) {
            ad = "E"
          } else {
            ad = "G"
          }
          var var1 = (dd + ad)
          array += var1
        }
      }
      (line._1, array)
    }
  }.filter(_._2.length > 0).flatMap {
    line => {
      var result: ArrayBuffer[(String, Int)] = ArrayBuffer[(String, Int)]()
      val values: ArrayBuffer[String] = line._2
      for (index <- 0 until (values.size - 1)) {
        result += new Tuple2[String, Int](values(index) + ":" + values(index + 1), 1)
      }
      result
    }
  }.coalesce(2).reduceByKey(_ + _)


  /*
    结果为：（(开始状态：结束状态)，统计值）
    SE:SG:2
    SG:SE:2
    SL:SE:4
    SE:SL:4
  */

  rdd.collect().foreach(a => println(a._1 + ":" + a._2))
  sc.stop()
}


/**
  * 传入广播的变量一定要线程安全和可序列化
  *
  * SimpleDateFormat 是线程不安全的。
  */
object MyDateParser extends Serializable {

  val df = new SimpleDateFormat("yyyy/MM/dd HH:mm")
  val df2 = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

  val lock = new ReentrantLock(false)

  def parse(dateStr: String): Date = {
    lock.lock()
    var date: Date = null
    try {
      if (dateStr.length == "yyyy/MM/dd HH:mm".length)
        date = df.parse(dateStr)
      else if (dateStr.length == "yyyy/MM/dd HH:mm:ss".length)
        date = df2.parse(dateStr)
      else
        date = df.parse(dateStr)
    } catch {
      case a => a.printStackTrace()
    } finally {
      lock.unlock()
    }
    date
  }
}
